package cn.ipanel.bigdata.boot

import cn.ipanel.bigdata.boot.Job.jobStart
import cn.ipanel.bigdata.boot.config.{ConfigKey, Configuration, Template}
import cn.ipanel.bigdata.boot.date.{Date, Day, Time}
import cn.ipanel.bigdata.boot.logger.Logger
import cn.ipanel.bigdata.dw.dim.phoenix.{T_TASK_STAT => S}
import cn.ipanel.bigdata.utils.Dictionary
import cn.ipanel.bigdata.utils.Util.isEmpty
import org.apache.commons.lang3.exception.ExceptionUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.DataFrame

import scala.reflect.ClassTag

/**
 * Author: lzz
 * Date: 2021/11/16 16:15
 */
trait Job extends Session with Serializable {

  lazy val PARAM_DATE: String = "date"
  lazy val PARAM_JOB_STAT: String = "saveJobStat"
  lazy val PARAM_UNKNOWN: String = Dictionary.UNKNOWN

  final lazy val _YYYYMMDD: String = getDateFromParam
  final lazy val _DAY: Day = if (isEmpty(_YYYYMMDD)) Date.asYesterday else Day(_YYYYMMDD)
  final lazy val _PAR_SAVE_JOB_STAT: String = getParam(PARAM_JOB_STAT)

  private[this] var bundle: Bundle = _

  private[Job] def setBundle(bundle: Bundle): Unit = this.bundle = bundle

  def getJobName: String = if (bundle == null) PARAM_UNKNOWN else bundle.getJobName
  def getParam(key: String): String = if (bundle == null) null else bundle.get(key)
  def getParams: String = if (bundle == null) null else bundle.getAll
  def getDateFromParam: String = getParam(PARAM_DATE)
  def getTemplate: Template = Configuration.getTemplate(getJobName)

  @Deprecated
  def sql(sql: String): DataFrame = spark.sql(sql)
  def broadcast[T : ClassTag](value: T): Broadcast[T] = spark.sparkContext.broadcast(value)

  def maxRuntime: Int = -1

  override def optCostumerConf: Map[String, String] = {null}

  /**
   * 任务准备阶段
   */
  def onPrepare(): Unit = {
    Logger.D(s"prepare Job[$getJobName][${bundle.getJobClasspath}]")
  }

  /**
   * 工作运行阶段
   */
  def onStartup(): Unit = {
    Logger.D(s"startup Job[$getJobName][${bundle.getJobClasspath}]")
  }

  /**
   * 工作成果检验阶段
   * @return 成果检验的结果
   */
  def onFruited(): Boolean = {
    Logger.D(s"fruited Job[$getJobName][${bundle.getJobClasspath}]")
    true
  }

  /**
   * 工作结束阶段
   */
  def onDestroy(): Unit = {
    Logger.D(s"destroy Job[$getJobName][${bundle.getJobClasspath}]")
  }

  def saveJobStat(errCode: Int, realtimeTaskRefresh: Boolean = false, tableName: String = Dictionary.TASK_EXECUTION_TABLE_NAME): Unit = {
    if (_PAR_SAVE_JOB_STAT == null) {
      if (Configuration.getParamOrElse(ConfigKey.SAVE_JOB_STAT, false)) {
        _saveStat(errCode, realtimeTaskRefresh, tableName)
      } else {
        Logger.I(s"param $PARAM_JOB_STAT is not set and config 'properties.param.${ConfigKey.SAVE_JOB_STAT}' with value 'false'(or not set), so job stat will not be saved")
      }
    } else if (_PAR_SAVE_JOB_STAT.toBoolean) {
      _saveStat(errCode, realtimeTaskRefresh, tableName)
    } else {
      Logger.I(s"param $PARAM_JOB_STAT is set 'false', so job stat will not be saved")
    }
  }

  private def _saveStat(errCode: Int, realtimeTaskRefresh: Boolean = false, tableName: String = Dictionary.TASK_EXECUTION_TABLE_NAME): Unit = {
    S.save(
      S.createSingleRowDataset(
        S.TaskStatTable(
          f_task_name = getJobName,
          f_table_name = tableName,
          f_app_id = spark.sparkContext.applicationId,
          f_start_time = jobStart.to(),
          f_end_time = if (!realtimeTaskRefresh) Time.now.to() else null,
          f_ref_time = if (realtimeTaskRefresh) Time.now.to() else null,
          f_param = getParams,
          f_error_code = errCode
        )
      )
    )
  }
}

object Job {

  private var jobStart: Time = Time.now

  def apply(b: Bundle): Job = {
    jobStart = Time.now
    var obj: Job = null
    if (b != null) {
      try {
        // Class.forName("cn.ipanel.bigdata.job." + "ods.Report")
        obj = Class.forName(b.getJobClasspath).newInstance().asInstanceOf[Job]
        obj.setBundle(b)
      } catch {
        case e: Exception =>
          Logger.E(
            s""" build object Job[${b.getJobName}][${b.getJobClasspath}] failed.
               | Because: ${ExceptionUtils.getStackTrace(e)}
               |""".stripMargin)
      }
    }
    obj
  }
}